package cn.witsky.rtp2ws.push.disruptor;

import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;

import cn.witsky.rtp2ws.Server;
import cn.witsky.rtp2ws.domain.Message;

public class EventCacheManage {
	public static BlockingQueue<Integer> eventDisruptorIndexQueue = null;
	private static Map<Integer, EventDisruptor> eventDisruptorMap = new ConcurrentHashMap<Integer, EventDisruptor>();// 缓存列表
	private static Map<String, Integer> callIdAndDisruptorIndexMap = null;


	/** 初始化方法 */
	public static void initialize() {
		if (Server.useNewThread) {
			return;
		}
		if (eventDisruptorMap.size() == 0) {
			synchronized (EventCacheManage.class) {
				if (eventDisruptorMap.size() == 0) {					
					eventDisruptorMap = new ConcurrentHashMap<Integer, EventDisruptor>();
					callIdAndDisruptorIndexMap = new ConcurrentHashMap<String, Integer>();
					for (int i= 1; i <= Server.THRED_COUNT; i++) {
						EventDisruptor cache = new EventDisruptor(i);
						eventDisruptorMap.put(i, cache);
					}
					int queueSize=Server.THRED_COUNT*100;
					eventDisruptorIndexQueue = new ArrayBlockingQueue(queueSize);					
					int j=1;					
					while(j<=queueSize) {
						int index=j%Server.THRED_COUNT;
						if(index==0) {
							eventDisruptorIndexQueue.add(Server.THRED_COUNT);
						}else {
							eventDisruptorIndexQueue.add(index);
						}

						j++;
					}
				}
			}
		} else {
			eventDisruptorMap.clear();
		}
	}

	public static void bindDisruptor(String callId) {
		if (callIdAndDisruptorIndexMap != null) {
			try {
				int index = eventDisruptorIndexQueue.take();
				callIdAndDisruptorIndexMap.put(callId, index);
				eventDisruptorIndexQueue.add(index);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}

	private static EventDisruptor getEventDisruptor(String callId) {
		Integer index = callIdAndDisruptorIndexMap.get(callId);
		if (index == null) {
			return null;
		} else {
			return eventDisruptorMap.get(index);
		}
	}

	public static void clearCall(String callId) {
		if (callIdAndDisruptorIndexMap != null) {
			callIdAndDisruptorIndexMap.remove(callId);
		}
	}

	/**
	 * 添加一个CacheObject对象
	 * 
	 * @param co CacheObject对象
	 */
	private static void offer(String callId, Message msg) {
		EventDisruptor disruptor = getEventDisruptor(callId);
		if (disruptor != null) {
			disruptor.getProducer().offer(msg);
		}

	}

	
	/**
	 * 添加一个CacheObject对象
	 * 
	 * @param co CacheObject对象
	 */
	public static void offer(boolean isCaller, String callId, int seq, byte[] bytes, long currentTimeMillis) {
		EventDisruptor disruptor = getEventDisruptor(callId);
		if (disruptor != null) {
			disruptor.getProducer().offer(isCaller, callId, seq, bytes, currentTimeMillis);
		}
	}
}